package cn.com.csg.smartenergy.mqtt;

import cn.com.csg.smartenergy.config.MqttConfig;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;

/**
 * mqtt client 抽象类
 *
 * @author ysguo
 * @date 2019-09-09
 */
@Data
@Slf4j
public abstract class AbstractMqttClient {

    protected MqttClient client;

    protected String clientId;

    /**
     * 用于断线重连时要订阅的Topics
     */
    protected Set<String> subTopics = new HashSet<>();

    /**
     * 创建连接
     *
     * @throws MqttException
     */
    public synchronized void connect(String clientId, AbstractCallBack callback) throws MqttException {
        log.info("connecting...");
        this.clientId = clientId;
        if (client == null) {
            client = new MqttClient(MqttConfig.host, clientId, new MemoryPersistence());
            client.setCallback(callback);
        }
        if (!client.isConnected()) {
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setMaxInflight(100000);
            options.setUserName(MqttConfig.username);
            options.setPassword(MqttConfig.password.toCharArray());
            options.setConnectionTimeout(15);
            options.setKeepAliveInterval(30);
            client.connect(options);
        }
    }

    /**
     * 关闭连接
     *
     * @throws MqttException
     */
    public void close() throws MqttException {
        if (Objects.isNull(this.client)) {
            return;
        }
        this.client.disconnect();
        this.client.close();
        log.info(String.format("ClientId: %s is closed", this.client.getClientId()));
    }

    /**
     * 发布消息
     *
     * @param topic 主题
     * @param message 消息
     * @throws MqttException
     */
    public void publish(String topic, MqttMessage message) throws MqttException {
        MqttTopic mqttTopic = this.client.getTopic(topic);
        log.info(String.format("ClientId: %s, Topic: %s", this.client.getClientId(), topic));
        log.info(String.format("ClientId: %s, message: %s", this.client.getClientId(), new String(message.getPayload(), StandardCharsets.UTF_8)));
        MqttDeliveryToken token = mqttTopic.publish(message);
//        token.waitForCompletion();
    }

    /**
     * 订阅消息
     *
     * @param topics 主题
     * @param qos 服务质量等级 0：almost once 1: at least once 2: exactly once
     * @throws MqttException
     */
    public void subscribe(String[] topics, int qos) throws MqttException {
        this.client.subscribe(topics, new int[]{qos});
    }

    /**
     * 订阅消息
     *
     * @param topic 主题
     * @param qos 服务质量等级 0：almost once 1: at least once 2: exactly once
     * @throws MqttException
     */
    public void subscribe(String topic, int qos) throws MqttException {
        this.client.subscribe(topic, qos);
    }

}
